/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.mqSender;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Struct;
import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Structs;
import cn.ac.iie.unpack.file.global.GlobalParas;
import java.util.logging.Level;

/**
 * @author Black-Laptop
 */
public class SendMQ {

    private static final Logger LOG = LoggerFactory.getLogger(SendMQ.class);
    private REConnection conn = new REConnection(GlobalParas.SEND_CONSUMER_NAMESERVER);

    RESendSessionBuilder builder;
    RESendSession session = null;

    private List<String> columns;
    private REFieldType[] types;

//    public SendMQ(String[] colNames, REFieldType[] types, Object[] datas) throws ConfigurationException {
//        this.colNames = Arrays.asList(colNames);
//        this.types = types;
//        addColumns(datas);
//    }
    public SendMQ(String[]columns,REFieldType[]types,String topic) {
        this.columns = Arrays.asList(columns);
        this.types = types;
        try {
            builder = (RESendSessionBuilder) conn.getSendSessionBuilder(topic);
        } catch (REConnectionException ex) {
            java.util.logging.Logger.getLogger(SendMQ.class.getName()).log(Level.SEVERE, null, ex);
        }
        //设置数据结构，参数为每一列的列名、列类型、是否可以为空。注意： 如果设置为不能为空，那么如果添加数据时没有添加该列数据，会报错。
        for (int i = 0; i < this.columns.size(); i++) {
            builder.addColumn(this.columns.get(i), types[i], true);
        }
        LOG.debug("colName: " + Arrays.toString(this.columns.toArray()) + "  type: " + Arrays.toString(types));
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
        LOG.info("topic: " + topic);
    }

    public void start() {
        int retryTime = 0;
        while (true) {
            try {
                session = (RESendSession) builder.build();
                session.start();
                LOG.info("session start successfully. retry time: " + retryTime);
                break;
            } catch (RESessionException | REConnectionException e) {
                retryTime++;
                LOG.error(ExceptionUtils.getFullStackTrace(e));
                stop();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    java.util.logging.Logger.getLogger(SendMQ.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }



    public void sendMessage( List<Object[]> datas){
        //生成builder，参数为topic，由具体业务决定

        /**
         * 这里是设置struct或者structs的格式。 首先需要addColumn
         * 其次需要在对应的类型里面setDetail，参数是：列名，子列类型map
         */
        //开始发送。
        //这里，DocSchemaName和UserDesc非必须，具体什么时候需要，由具体业务决定。
        //DocSchemaName设置一次后不会清空，每次flush都是同一个，如果需要改变，需要再次设置。
        //for (int r = 0; r < 10; ++r) {
        //添加数据。set完所有列后add即可添加一条数据。多条数据打一包发送。
        //各种set方法可以通过session.set补全来查看。set方法和REFieldType各类对应。
        //注意：尽量不要一条数据发送一次，这样会带来较大的额外开销
        //注意2：本地缓存数据量有大小限制，如果超过500M，会报错。
        //注意3：如果要用set（index，value）的方式设置数据，index从1开始。
        LOG.debug("datas size: " + datas.size());
        LOG.debug("type size: " + types.length);

        for (Object[] data : datas) {
            LOG.debug("send data: " + Arrays.toString(data));
            for (int i = 0; i < types.length; i++) {
                try {
                    if (data[i] == null || data[i].equals("null") || data[i].equals("\"null\"")) {
                        continue;
                    }
                    LOG.debug("column: " + columns.get(i) + " type:" + types[i] + " value: " + data[i]);
                    switch (types[i]) {
                        case String:
                            session.setString((String) columns.get(i), String.valueOf(data[i]));
                            break;
                        case Int:
                            session.setInt((String) columns.get(i), Integer.valueOf(data[i].toString()));
                            break;
                        case Long:
                            session.setLong((String) columns.get(i), Long.valueOf(data[i].toString()));
                            break;
                        case Float:
                            session.setFloat((String) columns.get(i), Float.valueOf(data[i].toString()));
                            break;
                        case Double:
                            session.setDouble((String) columns.get(i), Double.valueOf(data[i].toString()));
                            break;
                        case Binary:
                            session.setBinary((String) columns.get(i), data[i].toString().getBytes());
                            break;
                        case Doc:
                            session.setDoc((String) columns.get(i), (GenericRecord) data[i]);
                            break;
                        case Docs:
                            session.setDocs((String) columns.get(i), (GenericRecord[]) data[i]);
                            break;
                        case Binaries:
                            session.setBinaries((String) columns.get(i), (byte[][]) data[i]);
                            break;
                        case Doubles:
                            session.setDoubles((String) columns.get(i), (double[]) data[i]);
                            break;
                        case Floats:
                            session.setFloats((String) columns.get(i), (float[]) data[i]);
                            break;
                        case Longs:
                            session.setLongs((String) columns.get(i), (long[]) data[i]);
                            break;
                        case Ints:
                            session.setInts((String) columns.get(i), (int[]) data[i]);
                            break;
                        case Strings:
                            session.setStrings((String) columns.get(i), (String[]) data[i]);
                            break;
                        case Boolean:
                            session.setBoolean((String) columns.get(i), (boolean) data[i]);
                            break;
                        case Booleans:
                            session.setBooleans((String) columns.get(i), (boolean[]) data[i]);
                            break;
                        case Struct:
                            session.setStruct((String) columns.get(i), (Map) data[i]);
                            break;
                        case Structs:
                            session.setStructs((String) columns.get(i), (Map[]) data[i]);
                            break;
                        default:
                            LOG.error("There is not the type of " + types[i]);
                    }
                } catch (Exception ex) {
                    LOG.error(ExceptionUtils.getFullStackTrace(ex));
                }

            }
            try {
                session.add();
            } catch (RESessionException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
        }
        try {
            //添加完所有数据，flush即可发送。
            session.flush();
        } catch (RESessionException ex) {
            LOG.error(ExceptionUtils.getFullStackTrace(ex));
        } catch (REConnectionException ex) {
            LOG.error(ExceptionUtils.getFullStackTrace(ex));
        }
        LOG.info("Send Message " + (datas.size()));
    }

    public void stop() {
        try {
            session.shutdown();
        } catch (REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
        }
    }
}
